package cn.ac.iie.ulss.indexer.worker;

import cn.ac.iie.ulss.indexer.runenvs.GlobalParas;
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.MessageExt;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import org.apache.log4j.Logger;

public class RocketDataPuller implements Runnable {

    public static Logger log = Logger.getLogger(RocketDataPuller.class.getName());
    public String consumerGroupPrefix = null;
    public DefaultMQPushConsumer consumer = null;
    public List<String> oritTopics = null;
    public String tag = null;
    public ArrayBlockingQueue<MessageExt> inbuffer = null;
    private String consumerGroup = null;
    int min;
    int max;
    int callBack = 0;

    public RocketDataPuller(String consumerGroupPrefix, List<String> oriTopics, String tag, ArrayBlockingQueue buffer, int min, int max, int callback) {
        this.tag = tag;
        this.oritTopics = oriTopics;
        this.inbuffer = buffer;
        this.consumerGroupPrefix = consumerGroupPrefix;
        this.consumerGroup = (this.consumerGroupPrefix);
        this.consumer = new DefaultMQPushConsumer(this.consumerGroup);
        this.min = min;
        this.max = max;
        this.callBack = callback;
    }

    @Override
    public void run() {
        try {
            Thread.sleep(2000L);
        } catch (Exception e) {
            log.error(e, e);
        }
        try {
            this.consumer.setNamesrvAddr(GlobalParas.rocketNameServer);
            for (String s : this.oritTopics) {
                this.consumer.subscribe(GlobalParas.hostName + "_" + s, this.tag);
                log.info("subscribe message " + GlobalParas.hostName + "_" + s);
            }
            this.consumer.setClientIP(GlobalParas.ip);
            this.consumer.setInstanceName(this.consumerGroup + "ConsumeIns");
            this.consumer.setConsumeThreadMin(this.min);
            this.consumer.setConsumeThreadMax(this.max);
            this.consumer.setClientCallbackExecutorThreads(this.callBack);

            this.consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
            this.consumer.setConsumeMessageBatchMaxSize(GlobalParas.rocketConsumeMessageBatchMaxSize);
            this.consumer.setPullBatchSize(GlobalParas.rocketPullBatchSize);
            this.consumer.setPullThresholdForQueue(GlobalParas.pullThresholdForQueue);
            this.consumer.setConsumeConcurrentlyMaxSpan(GlobalParas.consumeConcurrentlyMaxSpan);
        } catch (Exception ex) {
            log.error(ex, ex);
        }
        try {
            this.consumer.registerMessageListener(new RocketdataHandler(this.inbuffer));
            this.consumer.start();
        } catch (Exception ex) {
            log.error(ex, ex);
        }
        log.info("consumer start ok for " + this.oritTopics + " ");
    }

    public void addTopic(String topic, String tag) {
        try {
            this.consumer.subscribe(topic, tag);
        } catch (Exception ex) {
            log.error(ex, ex);
        }
    }

    public void removeTopic(String topic) {
        try {
            this.consumer.unsubscribe(topic);
        } catch (Exception ex) {
            log.error(ex, ex);
        }
    }

    public void shutDownAll() {
        try {
            for (String s : this.oritTopics) {
                this.consumer.unsubscribe(s);
            }
            this.consumer.shutdown();
        } catch (Exception ex) {
            log.error(ex, ex);
        }
    }
}
